package com.chwl.framework.grpc.discover;

import com.alibaba.fastjson.JSON;
import com.chwl.framework.grpc.RemoteServer;
import com.chwl.framework.grpc.ServerContext;
import com.chwl.framework.grpc.etcd.EtcdBusienssServerKey;
import com.chwl.framework.grpc.etcd.EtcdService;
import com.chwl.framework.grpc.etcd.utils.NetWorkAddressUtils;
import com.chwl.framework.grpc.exception.NotFoundException;
import com.chwl.framework.grpc.exception.UserdefinedException;
import io.etcd.jetcd.KeyValue;
import io.etcd.jetcd.lease.LeaseKeepAliveResponse;
import io.etcd.jetcd.watch.WatchEvent;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import io.grpc.stub.StreamObserver;
import io.netty.util.concurrent.DefaultThreadFactory;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.collections4.MapUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 包名：com.anoyi.grpc.service.discover<br>
 * <p>功能:服务注册发现业务</p>
 *
 * @date:2018-07-14 下午9:25<br/>
 * @version:1.0 <br/>
 */
public class ServiceDiscoverService {
    private static Logger logger = LoggerFactory.getLogger(ServiceDiscoverService.class);
    private static Logger heartbeatLogger = LoggerFactory.getLogger("heartbeat");
    private ExecutorService serviceDiscoverExecutorService = new ThreadPoolExecutor(2, 2, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(), new DefaultThreadFactory("service-discover-service"));
    private ExecutorService serviceDiscoverCheckExecutorService = new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(), new DefaultThreadFactory("service-discover-check-service"));

    private static ServiceDiscoverService instance = new ServiceDiscoverService();
    ConcurrentHashMap<String, ConcurrentHashMap<String, ServerContext>> concurrentHashMap = new ConcurrentHashMap<String, ConcurrentHashMap<String, ServerContext>>();
    ConcurrentHashMap<String, AtomicInteger> deliveryService = new ConcurrentHashMap<String, AtomicInteger>();
    int MAX_VALUE = 1000000;

    private static List<String> serviceNames = new ArrayList<String>();

    public static ServiceDiscoverService getInstance() {
        return instance;
    }

    /**
     * 服务上报,服务发现
     *
     * @param grpcHeartbeat 是否开启grpc心跳上报(此配置打开后对内提供grpc服务)
     * @param port
     * @param serviceNames
     */
    public void heartbeat(int port, boolean grpcHeartbeat, List<String> serviceNames) {
        heartbeatLogger.info("service discover service,etcd client heartbeat start,port:{},grpcHeartbeat:{},serviceNames:{}", port, grpcHeartbeat, JSON.toJSONString(serviceNames));
        if (grpcHeartbeat) {
            //服务上报
            heartbeat(serviceNames, port, 10);
        }
        //初始化查询所有服务
        initAllService();
        //服务发现
        serviceDiscoverExecutorService.execute(() -> {
            watch(serviceNames);
        });
        //服务发现租约检查
        serviceDiscoverCheck(port, 10);
    }

    /**
     * 服务上报,服务发现
     *
     * @param port
     * @param serviceNames
     */
    public void heartbeat(int port, List<String> serviceNames) {
        heartbeat(port, true, serviceNames);
    }


    /**
     * 服务发现租约检查,如租约失效则做补偿(jetcd有BUG 当租约出现问题 直接close 且没有抛出来)
     *
     * @param port
     * @param rateTime
     */
    private void serviceDiscoverCheck(int port, int rateTime) {
        serviceDiscoverCheckExecutorService.execute(() -> {
            while (true) {
                boolean serviceDiscover = true;
                for (String serviceName : serviceNames) {
                    try {
                        String value = EtcdService.getInstance().get(serviceName);
                        if (StringUtils.isEmpty(value)) {
                            serviceDiscover = false;
                            break;
                        }
                    } catch (Exception e) {
                        heartbeatLogger.error("service discover check fail,serviceNames:{},error message:{}", JSON.toJSONString(serviceNames), e.getMessage(), e);
                    }
                }
                if (heartbeatLogger.isDebugEnabled()) {
                    heartbeatLogger.debug("service discover check successful,grpc on etcd is active,keepAlive fail,compensation start,serviceNames:{},", JSON.toJSONString(serviceNames));
                }
                if (!serviceDiscover) {
                    heartbeatLogger.info("service discover check successful,keepAlive fail,compensation start,serviceNames:{},", JSON.toJSONString(serviceNames));
                    heartbeat(serviceNames, port, rateTime);
                }
                try {
                    Thread.sleep(1000L);
                } catch (Exception e) {
                    heartbeatLogger.error("service discover check fail,serviceNames:{},error message:{}", JSON.toJSONString(serviceNames), e.getMessage(), e);
                }
            }
        });
    }

    /**
     * 服务心跳上报
     */
    public void heartbeat(List<String> services, int port, int rateTime) {
        heartbeatLogger.info("etcd client heartbeat start,serviceNames:{},rateTime:{}", JSON.toJSONString(services), rateTime);
        String ip = NetWorkAddressUtils.findLocalHostAddress();

        for (int i = 0; i < services.size(); i++) {
            String serviceName = services.get(i);
            //添加到集合 在关闭服务后统一清除当前节点
            String key = serviceName.startsWith(EtcdBusienssServerKey.RPC_SERVICE) ? serviceName : EtcdBusienssServerKey.RPC_SERVICE + serviceName + "/" + ip + ":" + port;
            if (!serviceNames.contains(key)) {
                serviceNames.add(key);
            }
        }
        heartbeatLogger.info("service discover service,etcd client heartbeat start,key:{}", JSON.toJSONString(serviceNames));
        serviceDiscoverExecutorService.execute(() -> {
            registry(ip, port, rateTime);
        });
    }

    private void registry(String ip, int port, int rateTime) {
        try {
            //注册服务
            List<Long> leaseIdList = serviceRegistry(ip, port, rateTime);
            //续约
            for (int i = 0; i < leaseIdList.size(); i++) {
                heartbeatLogger.info("service discover service,etcd client heartbeat start,leaseId:{}", leaseIdList.get(i));
                EtcdService.getInstance().getLeaseClient().keepAlive(leaseIdList.get(i), new StreamObserver<LeaseKeepAliveResponse>() {
                    @Override
                    public void onNext(LeaseKeepAliveResponse leaseKeepAliveResponse) {
                        logger.info("leaseKeepAliveResponse:{}", leaseKeepAliveResponse.getID() + ",ttl:{}", leaseKeepAliveResponse.getTTL());
                    }

                    @Override
                    public void onError(Throwable throwable) {

                    }

                    @Override
                    public void onCompleted() {

                    }
                });
                heartbeatLogger.info("service discover service,etcd client heartbeat finished,leaseId:{}", leaseIdList.get(i));
            }
            return;
        } catch (Exception e) {
            heartbeatLogger.error("service discover service,etcd client heartbeat fail,serviceNames:{},error message:{}", JSON.toJSONString(serviceNames), e.getMessage(), e);
            //服务续约异常则重新注册服务
            registry(ip, port, rateTime);
        }
    }

    private List<Long> serviceRegistry(String ip, int port, int rateTime) {
        List<Long> leaseIdList = new ArrayList<Long>(serviceNames.size());
        //注册服务
        try {
            for (int i = 0; i < serviceNames.size(); i++) {
                String key = serviceNames.get(i);
                //rateTime秒*2倍时间过期
                long leaseId = EtcdService.getInstance().put(key, ip + ":" + port, (long) (rateTime * 2));
                leaseIdList.add(leaseId);
                if (heartbeatLogger.isInfoEnabled()) {
                    heartbeatLogger.info("service discover service,etcd client heartbeat successful,key:{}", key);
                }
            }
        } catch (Exception e) {
            heartbeatLogger.error("service discover service,etcd client heartbeat fail,serviceNames:{},error message:{}", JSON.toJSONString(serviceNames), e.getMessage(), e);
        }
        return leaseIdList;
    }

    public void watch(List<String> serviceNames) {
        try {
            EtcdService.getInstance().watch(EtcdBusienssServerKey.RPC_SERVICE, (List<WatchEvent> watchEvents) -> {
                for (WatchEvent watchEvent : watchEvents) {
                    try {
                        String event = watchEvent.getEventType().toString();
                        KeyValue keyValue = watchEvent.getKeyValue();
                        RemoteServer server = makeRemoteServer(keyValue);
                        String serviceName = server.getServer();
                        if (CollectionUtils.isNotEmpty(serviceNames) && serviceNames.contains(serviceName)) {
                            //自身应用不用监听
                            continue;
                        }
                        if (heartbeatLogger.isInfoEnabled()) {
                            heartbeatLogger.info("service discover service,service watch,event:{},server:{},", event, JSON.toJSONString(server));
                        }
                        ConcurrentHashMap<String, ServerContext> hashMap = concurrentHashMap.get(serviceName);
                        hashMap = hashMap == null ? new ConcurrentHashMap<String, ServerContext>() : hashMap;

                        if (hashMap.containsKey(server.getHost()) && event.equals(WatchEvent.EventType.PUT.toString())) {
                            //当前服务节点已存在则不添加
                            continue;
                        }
                        heartbeatLogger.info("service discover service,service watch,event:{},server:{},", event, JSON.toJSONString(server));
                        if (event.equals(WatchEvent.EventType.DELETE.toString())) {
                            delete(serviceName, server.getHost());
                            continue;
                        }
                        if (event.equals(WatchEvent.EventType.PUT.toString())) {
                            hashMap.put(server.getHost(), makeServerContext(server));
                            concurrentHashMap.put(serviceName, hashMap);
                        }
                    } catch (Exception e) {
                        heartbeatLogger.error("service discover service,service watch fail,error message:{}", e.getMessage(), e);
                    }
                }
            });
        } catch (Exception e) {
            heartbeatLogger.error("service discover service,service watch fail,error message:{}", e.getMessage(), e);
        }
    }

    /**
     * 强制更新服务
     *
     * @param serviceName
     */
    private synchronized void forceFindAll(String serviceName) {
        ConcurrentHashMap<String, ServerContext> hashMap = concurrentHashMap.get(serviceName);
        if (MapUtils.isNotEmpty(hashMap)) {
            return;
        }

        try {
            List<KeyValue> keyValues = EtcdService.getInstance().findList(EtcdBusienssServerKey.RPC_SERVICE + serviceName);

            for (KeyValue keyValue : keyValues) {
                try {
                    RemoteServer server = makeRemoteServer(keyValue);

                    hashMap = concurrentHashMap.get(serviceName);
                    hashMap = hashMap == null ? new ConcurrentHashMap<String, ServerContext>() : hashMap;
                    hashMap.put(server.getHost(), makeServerContext(server));

                    concurrentHashMap.put(serviceName, hashMap);
                } catch (Exception e) {
                    logger.error("service discover service,force find all,error message:{}", e.getMessage(), e);
                }
            }
        } catch (Exception e) {
            logger.error("service discover service,force find all,error message:{}", e.getMessage(), e);
        }
    }


    /**
     * 通过服务名查询服务器
     *
     * @param serviceName
     */
    public ServerContext findServer(String serviceName, String ip) throws NotFoundException, UserdefinedException {
        //未指定IP则直接使用默认策略
        if (StringUtils.isEmpty(ip)) {
            return findServer(serviceName);
        }
        if (concurrentHashMap == null) {
            logger.error("service discover service,find server fail,all service is empty,serviceName:{}", serviceName);
            throw new NotFoundException("not found service server");
        }
        ConcurrentHashMap<String, ServerContext> current = concurrentHashMap.get(serviceName);
        if (MapUtils.isEmpty(current)) {
            logger.error("service discover service,find server fail,service is empty,serviceName:{}", serviceName);
            throw new NotFoundException("not found service server");
        }
        ServerContext context = current.get(ip);
        if (context == null) {
            logger.error("service discover service,find server fail,service server not found,serviceName:{},ip:{}", serviceName, ip);
            throw new NotFoundException("not found service server");
        }
        //已被执行关闭 则获取其他服务
        if (context.getChannel().isShutdown()) {
            return findServer(serviceName);
        }
        return context;
    }

    public ServerContext findServer(String serviceName) throws NotFoundException, UserdefinedException {
        return findServer(serviceName, 1);
    }

    private final int RETRY_MAX_COUNT = 5;

    /**
     * 通过服务名查询服务器
     *
     * @param serviceName
     */
    public ServerContext findServer(String serviceName, int retryCount) throws NotFoundException, UserdefinedException {
        if (MapUtils.isEmpty(concurrentHashMap)) {
            logger.error("service discover service,find server fail,all service is empty,serviceName:{}", serviceName);
            throw new NotFoundException("not found service server");
        }
        ConcurrentHashMap<String, ServerContext> current = concurrentHashMap.get(serviceName);
        if (MapUtils.isEmpty(current)) {
            if (retryCount < RETRY_MAX_COUNT) {
                retryCount++;
                forceFindAll(serviceName);
                return findServer(serviceName, retryCount);
            }
            logger.error("service discover service,find server fail,service is empty,serviceName:{}", serviceName);
            throw new NotFoundException("not found service server");
        }
        ArrayList<ServerContext> serverContexts = new ArrayList<ServerContext>(current.values());
        if (CollectionUtils.isEmpty(serverContexts)) {
            if (retryCount < RETRY_MAX_COUNT) {
                retryCount++;
                forceFindAll(serviceName);
                return findServer(serviceName, retryCount);
            }
            logger.error("service discover service,find server fail,service is empty,serviceName:{}", serviceName);
            throw new NotFoundException("not found service server");
        }
        try {
            int size = serverContexts.size();
            //TODO 通过取模获取服务器
            AtomicInteger atomicInteger = deliveryService.get(serviceName);
            if (atomicInteger != null) {
                int index = atomicInteger.incrementAndGet();
                if (index >= MAX_VALUE) {
                    atomicInteger.lazySet(1);
                }
                int deliveryIndex = index % size;
                ServerContext context = serverContexts.get(Math.min(deliveryIndex, size));
                //已被执行关闭 则获取其他服务
                if (context.getChannel().isShutdown()) {
                    return findServer(serviceName);
                }
                return context;
            }

            synchronized (this) {
                atomicInteger = deliveryService.get(serviceName);
                if (atomicInteger != null) {
                    int index = atomicInteger.incrementAndGet();
                    int deliveryIndex = index % size;
                    return serverContexts.get(Math.min(deliveryIndex, size));
                }

                atomicInteger = new AtomicInteger(1);
                deliveryService.put(serviceName, atomicInteger);
                //TODO 查询服务器 默认获取第一个
                ServerContext context = serverContexts.get(0);
                //已被执行关闭 则获取其他服务
                if (context.getChannel().isShutdown()) {
                    return findServer(serviceName);
                }
                return context;
            }
        } catch (Exception e) {
            logger.error("service discover service,find server fail,serviceName:{},error message:{}", serviceName, e.getMessage(), e);
            throw new UserdefinedException("service discover service,find server fail");
        }
    }

    public void delete(String serviceName, String host) {
        if (concurrentHashMap == null) {
            logger.error("service discover service,delete service fail,all service is empty,serviceName:{},host:{}", serviceName, host);
            return;
        }
        ConcurrentHashMap<String, ServerContext> current = concurrentHashMap.get(serviceName);
        if (current == null) {
            logger.error("service discover service,delete service fail,service is empty,serviceName:{},host:{}", serviceName, host);
            return;
        }
        try {
            current.get(host).getChannel().shutdown();
            current.remove(host);
            logger.info("service discover service,delete service successful,serviceName:{},host:{}", serviceName, host);
        } catch (Exception e) {
            logger.error("service discover service,delete service fail,serviceName:{},host:{},error message:{}", serviceName, host, e.getMessage(), e);

        }
    }

    /**
     * 初始化查找当前所有服务
     */
    public void initAllService() {

        List<KeyValue> keyValueList = EtcdService.getInstance().findList(EtcdBusienssServerKey.RPC_SERVICE);
        if (CollectionUtils.isEmpty(keyValueList)) {
            logger.error("service discover service,find all service,service is empty");
            return;
        }
        keyValueList.forEach(keyValue -> {
            RemoteServer remoteServer = makeRemoteServer(keyValue);
            ConcurrentHashMap<String, ServerContext> serverContexts = concurrentHashMap.get(remoteServer.getServer());
            serverContexts = serverContexts == null ? new ConcurrentHashMap<String, ServerContext>() : serverContexts;
            serverContexts.put(remoteServer.getHost(), makeServerContext(remoteServer));

            concurrentHashMap.put(remoteServer.getServer(), serverContexts);
        });
        logger.info("service discover service,find all service,services:{}", JSON.toJSONString(concurrentHashMap.keys()));
    }


    public ServerContext makeServerContext(RemoteServer remoteServer) {
        logger.info("service discover service,service watch successful,make serverContext,data:{}", JSON.toJSONString(remoteServer));
        ManagedChannel builder = ManagedChannelBuilder.forAddress(remoteServer.getHost(), remoteServer.getPort()).usePlaintext(true).build();
        return new ServerContext(builder, remoteServer.getHost(), remoteServer.getPort());
    }

    public RemoteServer makeRemoteServer(KeyValue value) {
        String key = value.getKey().toString(Charset.forName("utf-8"));
        key = key.substring(EtcdBusienssServerKey.RPC_SERVICE.length());
        String[] serviceAndAddress = key.split("/");
        String serviceName = serviceAndAddress[0];
        String addressStr = serviceAndAddress[1];

        int index = addressStr.lastIndexOf(":");
        String host = addressStr.substring(0, index);
        int port = Integer.parseInt(addressStr.substring(index + 1));
        return new RemoteServer(serviceName, host, port);
    }

    public void destroy() {
        try {
            serviceDiscoverExecutorService.shutdownNow();
            serviceNames.forEach(key -> {
                try {
                    EtcdService.getInstance().deleteByKey(key);
                    logger.info("etcd service delete key successful,key:{}", key);
                } catch (Exception e) {
                    logger.error("etcd service delete key fail,key:{},error message:{}", key, e.getMessage(), e);
                }
            });
        } catch (Exception e) {
            logger.error("service discover destroy fail,error message:{}", e.getMessage(), e);
        }
    }
}

